package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//import org.apache.flink.table.expressions.JsonObject;

/**
 * @author jiwei
 * @description
 * @date 2023/8/23 18:27
 */
public class TimeFrequencyAlignmentDemoV2 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id        BIGINT NOT NULL,\n" +     //采集信息id
                "    sid        STRING,\n" +    //水位传感器id
                "    ts   TIMESTAMP(3),\n" +    //数据采集时间
                "    flag   BOOLEAN,\n" +       //水位报警开关
                "    vc INT, \n" +                //采集到的水位高度
                "    watermark for ts as ts - interval '5' second\n" +                //采集到的水位高度
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '5', \n" +   //一次生成100万条数据
                "  'rows-per-second' = '1', \n" +   //每秒生成10000条数据
                "  'fields.sid.length' = '5', \n" +   //传感器id长度
                "  'fields.id.min' = '1', \n" +
                "  'fields.id.max' = '100', \n" +
                "  'fields.vc.min' = '10', \n" +
                "  'fields.vc.max' = '100'\n" +
                ")");




        Table table = tableEnv.sqlQuery("select concat('{" +
                "\"id\":" +
                "', cast(id as STRING), ',', '}') as ori_data, ts, flag, vc  from sourceTable");
        table.printSchema();

        tableEnv.toDataStream(table).print();
        env.execute();
    }

//    class FieldsToJsonStr extends ScalarFunction {
//        public String eval(@DataTypeHint(inputGroup = InputGroup.ANY)Object... args) {
//            JSONObject jsonObject = new JSONObject();
//
//
//        }
//    }



}
